package com.foreveross.fetl.application.impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.springframework.stereotype.Service;

import com.dayatang.domain.InstanceFactory;
import com.dayatang.querychannel.support.Page;
import com.foreveross.fetl.application.IAirLineCountApplication;
import com.foreveross.fetl.application.ISourceApplication;
import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.common.vo.AirLineCountVO;
import com.foreveross.fetl.core.domain.task.SourceEntity;
import com.foreveross.fetl.util.AirLineCountUtils;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.vo.SourceVO;

@Service("airLineCountApplication")
public class AirLineCountApplicationImpl implements IAirLineCountApplication {

	@Override
	public List<AirLineCountVO> findWeekList(Date date,String tableName) {
		if(null==date)
			throw new IllegalArgumentException("日期为空");
		long today=DateUtils.parse(DateUtils.format(new Date(), "yyyyMMdd"), "yyyyMMdd").getTime();
		long dateTimes=date.getTime();
		if(dateTimes>=today)
			throw new IllegalArgumentException("只能统计今天之前的数据");
		
		Long difference=today-dateTimes;
		long days=(difference/(1000*60*60*24));
		Calendar c=Calendar.getInstance();
		c.setTime(date);
		String dateStr=DateUtils.format(date, "yyyyMMdd");
		if(dateStr.compareTo("20130304")>=0)
			c.add(Calendar.DAY_OF_MONTH, -4);
		else if("20130301".equals(dateStr))
			c.add(Calendar.DAY_OF_MONTH, -1);
		else if("20130302".equals(dateStr))
			c.add(Calendar.DAY_OF_MONTH, -2);
		else if("20130303".equals(dateStr))
			c.add(Calendar.DAY_OF_MONTH, -3);
		
		int size=0;
		if(days>3){
			size=7;
		}else if(days==3){
			size=6;
		}else if(days==2){
			size=5;
		}else if(days==1){
			size=4;
		}
		List<AirLineCountVO> list=new ArrayList<AirLineCountVO>();
		for(int i=0;i<size ;i++){
			c.add(Calendar.DAY_OF_MONTH, 1);
			String rowkey="99999"+DateUtils.format(c.getTime(), "yyyyMMdd")+"999"+"999";
			Result res;
			try {
				res = HBaseHelper.getOneRecord(tableName, rowkey);
				if(null!=res){
					AirLineCountVO vo=AirLineCountUtils.getFormHbase(HBaseHelper.transformHBaseRowToMap(res));
					if(null!=vo){
						list.add(vo);
					}
				}
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		return list;
	}

	@Override
	public List<AirLineCountVO> findChannelList(Date date, String tableName) {
		if(null==date )
			throw new IllegalArgumentException("日期为空");
		SourceEntity entity =new SourceEntity();
		List<SourceEntity> listsource=entity.findlessCreateTime(date);
		String dateStr=DateUtils.format(date, "yyyyMMdd");
		List<AirLineCountVO> list=new ArrayList<AirLineCountVO>();
		for(SourceEntity source:listsource){
			String rowKey=source.getHbaseKey()+dateStr+"999999";
			Result res;
			try {
				res = HBaseHelper.getOneRecord(tableName, rowKey);
				if(null!=res){
					AirLineCountVO vo=AirLineCountUtils.getFormHbase(HBaseHelper.transformHBaseRowToMap(res));
					if(null!=vo){
						list.add(vo);	
					}
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		
		return list;
	}

	@Override
	public Page<AirLineCountVO> pageForChannel(Date date, String tableName,
			String channel, int page, int pageSize) {
		
		int i=0;
		if(null==date ){
			date=new Date();
			Date starDate=DateUtils.parse("2013-03-01", "yyyy-MM-dd");
			i=this.daysBetween(starDate,new Date())-1;
		}else
			i=1;
		Calendar c =Calendar.getInstance();
		c.setTime(date);
		c.add(Calendar.DAY_OF_MONTH, -1);
		
		if(null== channel || "".equals(channel))
			channel="99999";
		
		
	
		List<AirLineCountVO> list =new ArrayList<AirLineCountVO>();
		int startSize=(page-1)*pageSize;
		try {
			if(startSize!=0)
				c.add(Calendar.DAY_OF_MONTH, -startSize);
			for(int j=0;j<pageSize;j++){
				String rowKey=channel+DateUtils.format(c.getTime(), "yyyyMMdd")+"999999";
				Result res = HBaseHelper.getOneRecord(tableName, rowKey);
				if(null!=res){
					AirLineCountVO vo=AirLineCountUtils.getFormHbase(HBaseHelper.transformHBaseRowToMap(res));
					if(null!=vo)
						list.add(vo);
				}
				c.add(Calendar.DAY_OF_MONTH, -1);
			}
			
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		return new Page<AirLineCountVO>(page,i,pageSize,list);
	}

	@Override
	public Page<AirLineCountVO> pageForAirLine(Date date, String tableName,
			String channel, String fromCity, String toCity, int page,
			int pageSize) {
		int startSize=(page-1)*pageSize;
		int endSize=page*pageSize;
		int i=0;
		List<AirLineCountVO> list =new ArrayList<AirLineCountVO>();
		
		if(StringUtils.isBlank(channel) && null==date)
			return new Page<AirLineCountVO>(page,i,pageSize,list);
		
		if(StringUtils.isBlank(channel)){
			ISourceApplication sourceApplication=InstanceFactory.getInstance(ISourceApplication.class);
			List<SourceVO> listsource=sourceApplication.listAll();
			String dateStr=DateUtils.format(date, "yyyyMMdd");
			if(null==fromCity ||"".equals(fromCity))
				fromCity="999";
			if(null==toCity || "".equals(toCity))
				toCity="999";
			for(SourceVO vo:listsource){
				if(i>=startSize && i<endSize){
					String rowKey=vo.getHbaseKey()+dateStr+fromCity+toCity;
					try {
						Result res = HBaseHelper.getOneRecord(tableName, rowKey);
						AirLineCountVO countVO=AirLineCountUtils.getFormHbase(HBaseHelper.transformHBaseRowToMap(res));
						if(null!=countVO){
							list.add(countVO);
							i++;
						}
					} catch (IOException e) {
						e.printStackTrace();
					}
				}else{
					i++;	
				}
				
			}
			return new Page<AirLineCountVO>(page,i,pageSize,list);
		}
		
		if(null==date){
			if(null==fromCity ||"".equals(fromCity))
				fromCity="999";
			if(null==toCity || "".equals(toCity))
				toCity="999";
			Date startDate=DateUtils.parse("20130301", "yyyyMMdd");
			Date endDate=DateUtils.parse(DateUtils.format(new Date(), "yyyyMMdd"), "yyyyMMdd");
			Calendar cstart=Calendar.getInstance();
			cstart.setTime(startDate);
			Calendar cend=Calendar.getInstance();
			cend.setTime(endDate);
			while(cend.after(cstart)){
				cend.add(Calendar.DAY_OF_MONTH, -1);
				String rowKey=channel+DateUtils.format(cend.getTime(), "yyyyMMdd")+fromCity+toCity;
				if(i>=startSize && i<endSize){
					try {
						Result res = HBaseHelper.getOneRecord(tableName, rowKey);
						if(null!=res){
							AirLineCountVO countVO=AirLineCountUtils.getFormHbase(HBaseHelper.transformHBaseRowToMap(res));
							if(null!=countVO){
								list.add(countVO);
								i++;
							}
						}
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}else{
					i++;
				}
				
			}
			return new Page<AirLineCountVO>(page,i,pageSize,list);
		} 
		
		
		if(!StringUtils.isBlank(fromCity) && !StringUtils.isBlank(toCity)){
				String rowKey =channel +DateUtils.format(date, "yyyyMMdd")+fromCity+toCity;
				try {
					Result res = HBaseHelper.getOneRecord(tableName, rowKey);
					if(null!=res){
						AirLineCountVO countVO=AirLineCountUtils.getFormHbase(HBaseHelper.transformHBaseRowToMap(res));
						if(null!=countVO){
							list.add(countVO);
							i++;	
						}
					}
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				return new Page<AirLineCountVO>(page,i,pageSize,list);
			}
		
		if(!StringUtils.isBlank(fromCity)){
				String startRow=channel +DateUtils.format(date, "yyyyMMdd")+fromCity+"000";
				String StopRow=channel +DateUtils.format(date, "yyyyMMdd")+fromCity+"ZZZ";
				ResultScanner rs = null;
				try {
					rs=HBaseHelper.getRecords(tableName,startRow,StopRow);
					for (Result res : rs) {
						if(i>=startSize && i<endSize){
							AirLineCountVO countVO=AirLineCountUtils.getFormHbase(HBaseHelper.transformHBaseRowToMap(res));
							if(null!=countVO){
								list.add(countVO);
								i++;
							}
						}else{
							i++;
						}
						
					}
				} catch (IOException e) {
					e.printStackTrace();
				}
				return new Page<AirLineCountVO>(page,i,pageSize,list);

			}
		
		
				String startRow=channel +DateUtils.format(date, "yyyyMMdd")+"AAA"+"000";
				String StopRow=channel +DateUtils.format(date, "yyyyMMdd")+"ZZZ"+"ZZZ";
				ResultScanner rs = null;
				try {
					rs=HBaseHelper.getRecords(tableName,startRow,StopRow);
					for (Result res : rs) {
						if(i>=startSize && i<endSize){
							AirLineCountVO countVO=AirLineCountUtils.getFormHbase(HBaseHelper.transformHBaseRowToMap(res));
							if(null!=countVO){
								list.add(countVO);
								i++;
							}
						}else{
							i++;
						}
					}
				} catch (IOException e) {
					e.printStackTrace();
				}
			
		
		return new Page<AirLineCountVO>(page,i,pageSize,list);
	}
	
    /**   
     * 计算两个日期之间相差的天数   
     * @param date1   
     * @param date2   
     * @return   
     */    
    private  int daysBetween(Date date1,Date date2)     
    {     
        Calendar cal = Calendar.getInstance();     
        cal.setTime(date1);     
        long time1 = cal.getTimeInMillis();                  
        cal.setTime(date2);     
        long time2 = cal.getTimeInMillis();          
        long between_days=(time2-time1)/(1000*3600*24);     
             
       return Integer.parseInt(String.valueOf(between_days));            
    }    
}
